package com.maixy.rabbitmqdemo.peck;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * @ClassName PeckReceiver
 * @Description 失败重试有3种方法
 * 1.用redis或者mongo等第三方存储当前重试次数。
 * 2.在header中添加重试次数，并且使用channel.basicPublish() 方法重新将消息发送出去后将重试次数加1。
 * 3.使用spring-rabbit中自带的retry功能。
 *      3.1:重试并不是RabbitMQ重新发送了消息，仅仅是消费者内部进行了重试，换句话说就是重试跟mq没有任何关系；
 *        因此上述消费者代码不能添加try{}catch(){}，一旦捕获了异常，在自动ack模式下，就相当于消息正确处理了，
 *        消息直接被确认掉了，不会触发重试的；
 *
 *      3.2:retry只能在自动ack模式下使用。如果一定要在手动ack模式下使用retry功能，
 *      需保证消息能在有限次重试过程中可以重试成功，否则超过重试次数，又没办法执行ack或者nack，
 *      消息就会一直处于unack,并不会转发到死信队列
 *
 * @Date 2021/5/8 10:59
 * @Author maixy
 */
@Component
@Slf4j
public class PeckReceiver {

    /**
     * 用redis或者mongo等第三方存储当前重试次数。
     */
    @RabbitListener(queues = PeckRabbitConfig.PECK_QUEUE_1, ackMode = "MANUAL")
    @RabbitHandler
    public void receiver(String msg, Channel channel, Message message) throws Exception {
        try {
            log.info("consumer消费消息，消息内容是[{}]",msg);
            int i = 1/0;
            //Thread.sleep(1000);
            //在实际的项目开发中建议关闭消费者自动确认消息
            //参数2：是否开启批量应答，在实际的项目开发中建议关闭批量确认
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            log.error("consumer消费消息异常，消息内容是[{}]",msg);
            //拒绝当前消息，并把消息返回原来的队列
            //channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
        }
    }

    /**
     * 使用spring-rabbit中自带的retry功能
     */
    /*
    @RabbitListener(queues = PeckRabbitConfig.PECK_QUEUE_1)
    @RabbitHandler
    public void receiver(String msg, Channel channel, Message message) throws Exception {
        log.info("消费者消费消息：" + msg);
        int i = 1/0;
        //重试次数用完后还不成功，自动进入死信队列
    }*/

    /*@RabbitListener(queues = PeckRabbitConfig.PECK_QUEUE_2, ackMode = "MANUAL")
    @RabbitHandler
    public void receiver2(String msg, Channel channel, Message message) throws IOException {
        try {
            log.info("2消费者消费消息：" + msg);
            //确认消息消费成功
            //int i = 1/0;
            //Thread.sleep(1000);
            log.info("===================================2消费端获得消息:{}=============================================", LocalDateTime.now(ZoneId.systemDefault()));
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            log.error("2消费者消费消息异常，{}",msg);
            //拒绝当前消息，并把消息返回原来的队列
            //channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
        }
    }*/

    @RabbitListener(queues = PeckRabbitConfig.PECK_DEAD_QUEUE_1, ackMode = "MANUAL")
    @RabbitHandler
    public void userDeadLetterConsumer(String msg, Channel channel, Message message) {
        try {
            log.info("接收到死信消息:[{}]", msg);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            log.info("死信队列签收消息....消息路由键为:[{}]", message.getMessageProperties().getReceivedRoutingKey());
        } catch (IOException e) {
            log.error("死信队列消息签收失败", e);
        }
    }
}
